package com.study.chapter11;

import com.study.chapter05.source.ClickSource;
import com.study.entity.Event;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/19 20:25
 */
public class TableSqlTest {
    @SneakyThrows
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner((e, l) -> e.timestamp)
                );

        //获取表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //将数据流转换成表
        Table eventTable = tableEnv.fromDataStream(stream);

        //用SQL方式提取表的数据
        //法一：timestamp是关键字，所以要加引号
        Table visitTable = tableEnv.sqlQuery("select url,user, `timestamp` from " + eventTable);
        //法二：$()，需要引入对应的依赖
        Table userTab = eventTable.select($("url"), $("user"))
                .where($("user").isEqual("Alice"));

        //将表转换成数据流，并打印输出
        tableEnv.toDataStream(visitTable).print("visit");
        tableEnv.toDataStream(userTab).print("user");

        env.execute();
    }
}
